/*-------------------------------------------------------------------------
 *
 * hash.c
 *	  Margo Seltzer的哈希包在Postgres中的实现。
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  src/backend/access/hash/hash.c
 *
 * NOTES
 *	  本文件仅包含公共接口例程。
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/hash.h"
#include "access/hash_xlog.h"
#include "access/relscan.h"
#include "access/tableam.h"
#include "access/xloginsert.h"
#include "catalog/index.h"
#include "commands/progress.h"
#include "commands/vacuum.h"
#include "miscadmin.h"
#include "optimizer/plancat.h"
#include "pgstat.h"
#include "utils/builtins.h"
#include "utils/index_selfuncs.h"
#include "utils/rel.h"

/* 哈希构建的工作状态及其回调 */
typedef struct
{
	HSpool	   *spool;			/* 如果不使用溢出则为NULL */
	double		indtuples;		/* 接受的元组数量纳入索引 */
	Relation	heapRel;		/* 堆关系描述符 */
} HashBuildState;

static void fc_hashbuildCallback(Relation fc_index,
							  ItemPointer fc_tid,
							  Datum *fc_values,
							  bool *fc_isnull,
							  bool fc_tupleIsAlive,
							  void *fc_state);


/*
 * 哈希处理程序函数：返回带有访问方法参数和回调的IndexAmRoutine。
 */
Datum hashhandler(PG_FUNCTION_ARGS)
{
	IndexAmRoutine *fc_amroutine = makeNode(IndexAmRoutine);

	fc_amroutine->amstrategies = HTMaxStrategyNumber;
	fc_amroutine->amsupport = HASHNProcs;
	fc_amroutine->amoptsprocnum = HASHOPTIONS_PROC;
	fc_amroutine->amcanorder = false;
	fc_amroutine->amcanorderbyop = false;
	fc_amroutine->amcanbackward = true;
	fc_amroutine->amcanunique = false;
	fc_amroutine->amcanmulticol = false;
	fc_amroutine->amoptionalkey = false;
	fc_amroutine->amsearcharray = false;
	fc_amroutine->amsearchnulls = false;
	fc_amroutine->amstorage = false;
	fc_amroutine->amclusterable = false;
	fc_amroutine->ampredlocks = true;
	fc_amroutine->amcanparallel = false;
	fc_amroutine->amcaninclude = false;
	fc_amroutine->amusemaintenanceworkmem = false;
	fc_amroutine->amparallelvacuumoptions =
		VACUUM_OPTION_PARALLEL_BULKDEL;
	fc_amroutine->amkeytype = INT4OID;

	fc_amroutine->ambuild = hashbuild;
	fc_amroutine->ambuildempty = hashbuildempty;
	fc_amroutine->aminsert = hashinsert;
	fc_amroutine->ambulkdelete = hashbulkdelete;
	fc_amroutine->amvacuumcleanup = hashvacuumcleanup;
	fc_amroutine->amcanreturn = NULL;
	fc_amroutine->amcostestimate = hashcostestimate;
	fc_amroutine->amoptions = hashoptions;
	fc_amroutine->amproperty = NULL;
	fc_amroutine->ambuildphasename = NULL;
	fc_amroutine->amvalidate = hashvalidate;
	fc_amroutine->amadjustmembers = hashadjustmembers;
	fc_amroutine->ambeginscan = hashbeginscan;
	fc_amroutine->amrescan = hashrescan;
	fc_amroutine->amgettuple = hashgettuple;
	fc_amroutine->amgetbitmap = hashgetbitmap;
	fc_amroutine->amendscan = hashendscan;
	fc_amroutine->ammarkpos = NULL;
	fc_amroutine->amrestrpos = NULL;
	fc_amroutine->amestimateparallelscan = NULL;
	fc_amroutine->aminitparallelscan = NULL;
	fc_amroutine->amparallelrescan = NULL;

	PG_RETURN_POINTER(fc_amroutine);
}

/*
 *	hashbuild() -- 构建一个新的哈希索引。
 */
IndexBuildResult * hashbuild(Relation fc_heap, Relation fc_index, IndexInfo *fc_indexInfo)
{
	IndexBuildResult *fc_result;
	BlockNumber fc_relpages;
	double		fc_reltuples;
	double		fc_allvisfrac;
	uint32		fc_num_buckets;
	long		fc_sort_threshold;
	HashBuildState fc_buildstate;

	/*
	 * 我们希望每个索引关系恰好调用一次。如果不是，
	 * 那将是大麻烦。
	 */
	if (RelationGetNumberOfBlocks(fc_index) != 0)
		elog(ERROR, "index \"%s\" already contains data",
			 RelationGetRelationName(fc_index));

	/* 估计当前表中存在的行数 */
	estimate_rel_size(fc_heap, NULL, &fc_relpages, &fc_reltuples, &fc_allvisfrac);

	/* 初始化哈希索引元数据页和初始桶 */
	fc_num_buckets = _hash_init(fc_index, fc_reltuples, MAIN_FORKNUM);

	/*
	 * 如果我们只是按扫描顺序将元组插入索引中，那么
	 * （假设它们的哈希代码是相当随机的），就不会有局部性
	 * 的索引访问，如果索引大于可用RAM，
	 * 我们将会严重影响性能。为防止这种情况，我们可以按 
	 * （预期的）桶编号对元组进行排序。然而，当索引适合 
	 * RAM 时，这样的排序是无用的开销。当初始索引大小超过 
	 * maintenance_work_mem，或可用于索引的缓冲区数量（以较小者为准）时，我们选择排序。 
	 * （通过缓冲区数量限制应该减少PG缓冲区和内核缓冲区之间的争用，
	 * 这似乎即使没有物理I/O结果也很有用。
	 * 通过maintenance_work_mem限制的好处是允许轻松测试排序 
	 * 代码路径，并且可能对DBA作为额外的控制选项有用。）
	 *
	 * 注意：如果一个桶的大小曾经与一页不同，则需要调整此测试。
	 * 此外，“初始索引大小”计算不包括元页和第一个位图页。
	 */
	fc_sort_threshold = (maintenance_work_mem * 1024L) / BLCKSZ;
	if (fc_index->rd_rel->relpersistence != RELPERSISTENCE_TEMP)
		fc_sort_threshold = Min(fc_sort_threshold, NBuffers);
	else
		fc_sort_threshold = Min(fc_sort_threshold, NLocBuffer);

	if (fc_num_buckets >= (uint32) fc_sort_threshold)
		fc_buildstate.spool = _h_spoolinit(fc_heap, fc_index, fc_num_buckets);
	else
		fc_buildstate.spool = NULL;

	/* 准备构建索引 */
	fc_buildstate.indtuples = 0;
	fc_buildstate.heapRel = fc_heap;

	/* 执行堆扫描 */
	fc_reltuples = table_index_build_scan(fc_heap, fc_index, fc_indexInfo, true, true,
									   fc_hashbuildCallback,
									   (void *) &fc_buildstate, NULL);
	pgstat_progress_update_param(PROGRESS_CREATEIDX_TUPLES_TOTAL,
								 fc_buildstate.indtuples);

	if (fc_buildstate.spool)
	{
		/* 对元组进行排序并将其插入索引中 */
		_h_indexbuild(fc_buildstate.spool, fc_buildstate.heapRel);
		_h_spooldestroy(fc_buildstate.spool);
	}

	/*
	 * 返回统计信息
	 */
	fc_result = (IndexBuildResult *) palloc(sizeof(IndexBuildResult));

	fc_result->heap_tuples = fc_reltuples;
	fc_result->index_tuples = fc_buildstate.indtuples;

	return fc_result;
}

/*
 *	hashbuildempty() -- 在初始化分支中构建一个空的哈希索引
 */
void hashbuildempty(Relation fc_index)
{
	_hash_init(fc_index, 0, INIT_FORKNUM);
}

/*
 * 每元组的回调，用于table_index_build_scan
 */
static void fc_hashbuildCallback(Relation fc_index,
				  ItemPointer fc_tid,
				  Datum *fc_values,
				  bool *fc_isnull,
				  bool fc_tupleIsAlive,
				  void *fc_state)
{
	HashBuildState *fc_buildstate = (HashBuildState *) fc_state;
	Datum		fc_index_values[1];
	bool		fc_index_isnull[1];
	IndexTuple	fc_itup;

	/* 将数据转换为哈希键；失败时不插入任何内容 */
	if (!_hash_convert_tuple(fc_index,
							 fc_values, fc_isnull,
							 fc_index_values, fc_index_isnull))
		return;

	/* 要么将元组进行排序，要么直接放入索引 */
	if (fc_buildstate->spool)
		_h_spool(fc_buildstate->spool, fc_tid, fc_index_values, fc_index_isnull);
	else
	{
		/* 形成一个索引元组并指向堆元组 */
		fc_itup = index_form_tuple(RelationGetDescr(fc_index),
								fc_index_values, fc_index_isnull);
		fc_itup->t_tid = *fc_tid;
		_hash_doinsert(fc_index, fc_itup, fc_buildstate->heapRel);
		pfree(fc_itup);
	}

	fc_buildstate->indtuples += 1;
}

/*
 *	hashinsert() -- 将索引元组插入哈希表中。
 *
 *	根据堆元组的键进行哈希，形成一个带有哈希代码的索引元组。
 *	找到新元组的适当位置，并将其放在那里。
 */
bool hashinsert(Relation fc_rel, Datum *fc_values, bool *fc_isnull,
		   ItemPointer fc_ht_ctid, Relation fc_heapRel,
		   IndexUniqueCheck fc_checkUnique,
		   bool fc_indexUnchanged,
		   IndexInfo *fc_indexInfo)
{
	Datum		fc_index_values[1];
	bool		fc_index_isnull[1];
	IndexTuple	fc_itup;

	/* 将数据转换为哈希键；失败时不插入任何内容 */
	if (!_hash_convert_tuple(fc_rel,
							 fc_values, fc_isnull,
							 fc_index_values, fc_index_isnull))
		return false;

	/* 形成一个索引元组并指向堆元组 */
	fc_itup = index_form_tuple(RelationGetDescr(fc_rel), fc_index_values, fc_index_isnull);
	fc_itup->t_tid = *fc_ht_ctid;

	_hash_doinsert(fc_rel, fc_itup, fc_heapRel);

	pfree(fc_itup);

	return false;
}


/*
 *	hashgettuple() -- 获取扫描中的下一个元组。
 */
bool hashgettuple(IndexScanDesc fc_scan, ScanDirection fc_dir)
{
	HashScanOpaque fc_so = (HashScanOpaque) fc_scan->opaque;
	bool		fc_res;

	/* 哈希索引总是有损的，因为我们只存储哈希代码 */
	fc_scan->xs_recheck = true;

	/*
	 * 如果我们已经初始化了这个扫描，我们可以在
	 * 适当的方向上向前推进。如果我们还没有这样做，
	 * 我们调用一个例程以获取扫描中的第一个项。
	 */
	if (!HashScanPosIsValid(fc_so->currPos))
		fc_res = _hash_first(fc_scan, fc_dir);
	else
	{
		/*
		 * 检查是否应该丢弃先前获取的元组。
		 */
		if (fc_scan->kill_prior_tuple)
		{
			/*
			 * 是的，所以请记住它以供后用。（我们将在离开索引页面或扫描结束时
			 * 一次处理所有这样的元组。）
			 * 如果调用者反转索引扫描方向，相同的项可能会被多次输入。
			 * 但是，我们不检测此情况；相反，我们只是忘记任何多余的条目。
			 */
			if (fc_so->killedItems == NULL)
				fc_so->killedItems = (int *)
					palloc(MaxIndexTuplesPerPage * sizeof(int));

			if (fc_so->numKilled < MaxIndexTuplesPerPage)
				fc_so->killedItems[fc_so->numKilled++] = fc_so->currPos.itemIndex;
		}

		/*
		 * 现在继续扫描。
		 */
		fc_res = _hash_next(fc_scan, fc_dir);
	}

	return fc_res;
}


/*
 *	hashgetbitmap() -- 一次获取所有元组
 */
int64 hashgetbitmap(IndexScanDesc fc_scan, TIDBitmap *fc_tbm)
{
	HashScanOpaque fc_so = (HashScanOpaque) fc_scan->opaque;
	bool		fc_res;
	int64		fc_ntids = 0;
	HashScanPosItem *fc_currItem;

	fc_res = _hash_first(fc_scan, ForwardScanDirection);

	while (fc_res)
	{
		fc_currItem = &fc_so->currPos.items[fc_so->currPos.itemIndex];

		/*
		 * _hash_first 和 _hash_next 处理消除死索引条目
		 * 每当 scan->ignore_killed_tuples 为真时。因此，除了将结果添加到 TIDBitmap 中
		 * 之外没有其他操作。
		 */
		tbm_add_tuples(fc_tbm, &(fc_currItem->heapTid), 1, true);
		fc_ntids++;

		fc_res = _hash_next(fc_scan, ForwardScanDirection);
	}

	return fc_ntids;
}


/*
 *	hashbeginscan() -- 在哈希索引上开始扫描
 */
IndexScanDesc hashbeginscan(Relation fc_rel, int fc_nkeys, int fc_norderbys)
{
	IndexScanDesc fc_scan;
	HashScanOpaque fc_so;

	/* 不允许使用 order by 操作符 */
	Assert(fc_norderbys == 0);

	fc_scan = RelationGetIndexScan(fc_rel, fc_nkeys, fc_norderbys);

	fc_so = (HashScanOpaque) palloc(sizeof(HashScanOpaqueData));
	HashScanPosInvalidate(fc_so->currPos);
	fc_so->hashso_bucket_buf = InvalidBuffer;
	fc_so->hashso_split_bucket_buf = InvalidBuffer;

	fc_so->hashso_buc_populated = false;
	fc_so->hashso_buc_split = false;

	fc_so->killedItems = NULL;
	fc_so->numKilled = 0;

	fc_scan->opaque = fc_so;

	return fc_scan;
}

/*
 *	hashrescan() -- 重新扫描索引关系
 */
void hashrescan(IndexScanDesc fc_scan, ScanKey fc_scankey, int fc_nscankeys,
		   ScanKey fc_orderbys, int fc_norderbys)
{
	HashScanOpaque fc_so = (HashScanOpaque) fc_scan->opaque;
	Relation	fc_rel = fc_scan->indexRelation;

	if (HashScanPosIsValid(fc_so->currPos))
	{
		/* 在离开当前页面之前，处理任何已删除的项 */
		if (fc_so->numKilled > 0)
			_hash_kill_items(fc_scan);
	}

	_hash_dropscanbuf(fc_rel, fc_so);

	/* 将位置设置为无效（这将导致调用 _hash_first） */
	HashScanPosInvalidate(fc_so->currPos);

	/* 更新扫描关键字，如果提供了一个新的关键字 */
	if (fc_scankey && fc_scan->numberOfKeys > 0)
	{
		memmove(fc_scan->keyData,
				fc_scankey,
				fc_scan->numberOfKeys * sizeof(ScanKeyData));
	}

	fc_so->hashso_buc_populated = false;
	fc_so->hashso_buc_split = false;
}

/*
 *	hashendscan() -- 关闭扫描
 */
void hashendscan(IndexScanDesc fc_scan)
{
	HashScanOpaque fc_so = (HashScanOpaque) fc_scan->opaque;
	Relation	fc_rel = fc_scan->indexRelation;

	if (HashScanPosIsValid(fc_so->currPos))
	{
		/* 在离开当前页面之前，处理任何已删除的项 */
		if (fc_so->numKilled > 0)
			_hash_kill_items(fc_scan);
	}

	_hash_dropscanbuf(fc_rel, fc_so);

	if (fc_so->killedItems != NULL)
		pfree(fc_so->killedItems);
	pfree(fc_so);
	fc_scan->opaque = NULL;
}

/*
 * 批量删除指向一组堆元组的所有索引条目。
 * 目标元组的集合通过回调例程指定，该例程指示任何给定的堆元组（由 ItemPointer 标识）是否正在被删除。
 *
 * 此函数还删除通过分裂移到其他桶的元组。
 *
 * 结果：一个 palloc'd 结构，包含用于 VACUUM 显示的统计信息。
 */
IndexBulkDeleteResult * hashbulkdelete(IndexVacuumInfo *fc_info, IndexBulkDeleteResult *fc_stats,
			   IndexBulkDeleteCallback fc_callback, void *fc_callback_state)
{
	Relation	fc_rel = fc_info->index;
	double		fc_tuples_removed;
	double		fc_num_index_tuples;
	double		fc_orig_ntuples;
	Bucket		fc_orig_maxbucket;
	Bucket		fc_cur_maxbucket;
	Bucket		fc_cur_bucket;
	Buffer		fc_metabuf = InvalidBuffer;
	HashMetaPage fc_metap;
	HashMetaPage fc_cachedmetap;

	fc_tuples_removed = 0;
	fc_num_index_tuples = 0;

	/*
	 * 我们需要一个元数据页面的副本，以便可以使用其 hashm_spares[]
	 * 值来计算桶页面地址，但缓存副本就足够了。
	 * （如果不够，我们将在进一步下探时检测到并根据需要刷新缓存。）
	 */
	fc_cachedmetap = _hash_getcachedmetap(fc_rel, &fc_metabuf, false);
	Assert(fc_cachedmetap != NULL);

	fc_orig_maxbucket = fc_cachedmetap->hashm_maxbucket;
	fc_orig_ntuples = fc_cachedmetap->hashm_ntuples;

	/* 扫描我们知道存在的桶 */
	fc_cur_bucket = 0;
	fc_cur_maxbucket = fc_orig_maxbucket;

loop_top:
	while (fc_cur_bucket <= fc_cur_maxbucket)
	{
		BlockNumber fc_bucket_blkno;
		BlockNumber fc_blkno;
		Buffer		fc_bucket_buf;
		Buffer		fc_buf;
		HashPageOpaque fc_bucket_opaque;
		Page		fc_page;
		bool		fc_split_cleanup = false;

		/* 获取桶的起始页面的地址 */
		fc_bucket_blkno = BUCKET_TO_BLKNO(fc_cachedmetap, fc_cur_bucket);

		fc_blkno = fc_bucket_blkno;

		/*
		 * 我们需要在主桶页面上获取清理锁，以便在删除死元组之前
		 * 等待并发扫描结束。
		 */
		fc_buf = ReadBufferExtended(fc_rel, MAIN_FORKNUM, fc_blkno, RBM_NORMAL, fc_info->strategy);
		LockBufferForCleanup(fc_buf);
		_hash_checkpage(fc_rel, fc_buf, LH_BUCKET_PAGE);

		fc_page = BufferGetPage(fc_buf);
		fc_bucket_opaque = HashPageGetOpaque(fc_page);

		/*
		 * 如果桶包含通过分裂移动的元组，则我们需要
		 * 删除这些元组。如果桶的分裂操作尚未完成，我们无法删除这些元组，因为扫描需要它们。
		 */
		if (!H_BUCKET_BEING_SPLIT(fc_bucket_opaque) &&
			H_NEEDS_SPLIT_CLEANUP(fc_bucket_opaque))
		{
			fc_split_cleanup = true;

			/*
			 * 自从我们上次在元数据页面上持有锁以来，这个桶可能已经被分裂。
			 * 如果是这样，hashm_maxbucket、hashm_highmask 和
			 * hashm_lowmask 可能已经过时，以至于导致我们无法删除
			 * 最近分裂留下的元组。为了防止这种情况，现在在目标桶的主页面被锁定
			 * （因此不能再分裂）后，检查是否需要
			 * 更新我们缓存的元数据页面数据。
			 */
			Assert(fc_bucket_opaque->hasho_prevblkno != InvalidBlockNumber);
			if (fc_bucket_opaque->hasho_prevblkno > fc_cachedmetap->hashm_maxbucket)
			{
				fc_cachedmetap = _hash_getcachedmetap(fc_rel, &fc_metabuf, true);
				Assert(fc_cachedmetap != NULL);
			}
		}

		fc_bucket_buf = fc_buf;

		hashbucketcleanup(fc_rel, fc_cur_bucket, fc_bucket_buf, fc_blkno, fc_info->strategy,
						  fc_cachedmetap->hashm_maxbucket,
						  fc_cachedmetap->hashm_highmask,
						  fc_cachedmetap->hashm_lowmask, &fc_tuples_removed,
						  &fc_num_index_tuples, fc_split_cleanup,
						  fc_callback, fc_callback_state);

		_hash_dropbuf(fc_rel, fc_bucket_buf);

		/* 进到下一个桶 */
		fc_cur_bucket++;
	}

	if (BufferIsInvalid(fc_metabuf))
		fc_metabuf = _hash_getbuf(fc_rel, HASH_METAPAGE, HASH_NOLOCK, LH_META_PAGE);

	/* 写锁定元数据页面并检查自我们开始以来是否发生了分裂 */
	LockBuffer(fc_metabuf, BUFFER_LOCK_EXCLUSIVE);
	fc_metap = HashPageGetMeta(BufferGetPage(fc_metabuf));

	if (fc_cur_maxbucket != fc_metap->hashm_maxbucket)
	{
		/* 已经发生了分裂，因此处理额外的桶 */
		LockBuffer(fc_metabuf, BUFFER_LOCK_UNLOCK);
		fc_cachedmetap = _hash_getcachedmetap(fc_rel, &fc_metabuf, true);
		Assert(fc_cachedmetap != NULL);
		fc_cur_maxbucket = fc_cachedmetap->hashm_maxbucket;
		goto loop_top;
	}

	/* 好吧，我们真的完成了。更新元数据页面中的元组计数。 */
	START_CRIT_SECTION();

	if (fc_orig_maxbucket == fc_metap->hashm_maxbucket &&
		fc_orig_ntuples == fc_metap->hashm_ntuples)
	{
		/*
		 * 自扫描开始以来没有人进行过分割或插入，所以
		 * 相信我们的计数是绝对准确的。
		 */
		fc_metap->hashm_ntuples = fc_num_index_tuples;
	}
	else
	{
		/*
		 * 否则，我们的计数是不可靠的，因为我们可能在分割桶中
		 * 进行了双重扫描元组。继续使用推测法。
		 * （注意：我们仍然返回estimated_count = false，因为使用这个
		 * 计数比根本不更新reltuples要好。）
		 */
		if (fc_metap->hashm_ntuples > fc_tuples_removed)
			fc_metap->hashm_ntuples -= fc_tuples_removed;
		else
			fc_metap->hashm_ntuples = 0;
		fc_num_index_tuples = fc_metap->hashm_ntuples;
	}

	MarkBufferDirty(fc_metabuf);

	/* XLOG 相关内容 */
	if (RelationNeedsWAL(fc_rel))
	{
		xl_hash_update_meta_page fc_xlrec;
		XLogRecPtr	fc_recptr;

		fc_xlrec.ntuples = fc_metap->hashm_ntuples;

		XLogBeginInsert();
		XLogRegisterData((char *) &fc_xlrec, SizeOfHashUpdateMetaPage);

		XLogRegisterBuffer(0, fc_metabuf, REGBUF_STANDARD);

		fc_recptr = XLogInsert(RM_HASH_ID, XLOG_HASH_UPDATE_META_PAGE);
		PageSetLSN(BufferGetPage(fc_metabuf), fc_recptr);
	}

	END_CRIT_SECTION();

	_hash_relbuf(fc_rel, fc_metabuf);

	/* 返回统计信息 */
	if (fc_stats == NULL)
		fc_stats = (IndexBulkDeleteResult *) palloc0(sizeof(IndexBulkDeleteResult));
	fc_stats->estimated_count = false;
	fc_stats->num_index_tuples = fc_num_index_tuples;
	fc_stats->tuples_removed += fc_tuples_removed;
	/* hashvacuumcleanup将填充num_pages */

	return fc_stats;
}

/*
 * 后VACUUM清理。
 *
 * 结果：一个palloc'd结构包含VACUUM显示的统计信息。
 */
IndexBulkDeleteResult * hashvacuumcleanup(IndexVacuumInfo *fc_info, IndexBulkDeleteResult *fc_stats)
{
	Relation	fc_rel = fc_info->index;
	BlockNumber fc_num_pages;

	/* 如果没有调用hashbulkdelete，则返回NULL表示没有变化 */
	/* 注意：这也涵盖了仅分析的情况 */
	if (fc_stats == NULL)
		return NULL;

	/* 更新统计信息 */
	fc_num_pages = RelationGetNumberOfBlocks(fc_rel);
	fc_stats->num_pages = fc_num_pages;

	return fc_stats;
}

/*
 * 辅助函数，用于从桶中删除索引条目。
 *
 * 此函数期望调用者已在
 * 主桶页面上获得清理锁，并将在
 * 主桶页面上再次持有写锁。 该锁不一定会持续持有，
 * 因为我们将在访问溢出页面时释放它。
 *
 * 当我们第一次进入该
 * 函数时，主桶页面上持有的清理锁确保没有任何并发扫描正在进行，
 * 但一旦释放该锁，可能就会有并发扫描。如果这些扫描在我们的清理
 * 扫描之前完成，它们可能会在我们删除元组之前看到它，并且只有在
 * VACUUM 完成并且 TID 已被回收用于无关元组后才会唤醒。为了避免
 * 这种灾难，我们通过在释放前一页的锁之前锁定桶链中的下一页
 * 来防止扫描超过我们的清理扫描。（这种类型的锁链并不理想，
 * 所以我们可能希望在某个时候寻找更好的解决方案。）
 *
 * 我们需要保持主桶的引脚，以确保没有并发的
 * 分割可以开始。
 */
void hashbucketcleanup(Relation fc_rel, Bucket fc_cur_bucket, Buffer fc_bucket_buf,
				  BlockNumber fc_bucket_blkno, BufferAccessStrategy fc_bstrategy,
				  uint32 fc_maxbucket, uint32 fc_highmask, uint32 fc_lowmask,
				  double *fc_tuples_removed, double *fc_num_index_tuples,
				  bool fc_split_cleanup,
				  IndexBulkDeleteCallback fc_callback, void *fc_callback_state)
{
	BlockNumber fc_blkno;
	Buffer		fc_buf;
	Bucket		fc_new_bucket PG_USED_FOR_ASSERTS_ONLY = InvalidBucket;
	bool		fc_bucket_dirty = false;

	fc_blkno = fc_bucket_blkno;
	fc_buf = fc_bucket_buf;

	if (fc_split_cleanup)
		fc_new_bucket = _hash_get_newbucket_from_oldbucket(fc_rel, fc_cur_bucket,
														fc_lowmask, fc_maxbucket);

	/* 扫描桶中的每一页 */
	for (;;)
	{
		HashPageOpaque fc_opaque;
		OffsetNumber fc_offno;
		OffsetNumber fc_maxoffno;
		Buffer		fc_next_buf;
		Page		fc_page;
		OffsetNumber fc_deletable[MaxOffsetNumber];
		int			fc_ndeletable = 0;
		bool		fc_retain_pin = false;
		bool		fc_clear_dead_marking = false;

		vacuum_delay_point();

		fc_page = BufferGetPage(fc_buf);
		fc_opaque = HashPageGetOpaque(fc_page);

		/* 扫描页中的每个元组 */
		fc_maxoffno = PageGetMaxOffsetNumber(fc_page);
		for (fc_offno = FirstOffsetNumber;
			 fc_offno <= fc_maxoffno;
			 fc_offno = OffsetNumberNext(fc_offno))
		{
			ItemPointer fc_htup;
			IndexTuple	fc_itup;
			Bucket		fc_bucket;
			bool		fc_kill_tuple = false;

			fc_itup = (IndexTuple) PageGetItem(fc_page,
											PageGetItemId(fc_page, fc_offno));
			fc_htup = &(fc_itup->t_tid);

			/*
			 * 为了删除死元组，我们严格依赖于回调函数的结果。
			 * 参见btvacuumpage获取详细原因。
			 */
			if (fc_callback && fc_callback(fc_htup, fc_callback_state))
			{
				fc_kill_tuple = true;
				if (fc_tuples_removed)
					*fc_tuples_removed += 1;
			}
			else if (fc_split_cleanup)
			{
				/* 删除被分割移动的元组。 */
				fc_bucket = _hash_hashkey2bucket(_hash_get_indextuple_hashkey(fc_itup),
											  fc_maxbucket,
											  fc_highmask,
											  fc_lowmask);
				/* 标记项目以供删除 */
				if (fc_bucket != fc_cur_bucket)
				{
					/*
					 * 我们期望元组要么属于当前桶，要么属于
					 * new_bucket。这是通过不允许来自包含垃圾的桶
					 * 进一步分割来确保的。请参见_hash_expandtable中的评论。
					 */
					Assert(fc_bucket == fc_new_bucket);
					fc_kill_tuple = true;
				}
			}

			if (fc_kill_tuple)
			{
				/* 标记项目以供删除 */
				fc_deletable[fc_ndeletable++] = fc_offno;
			}
			else
			{
				/* 我们继续保留它，因此计数它 */
				if (fc_num_index_tuples)
					*fc_num_index_tuples += 1;
			}
		}

		/* 在桶扫描结束之前保持主桶页面的引脚 */
		if (fc_blkno == fc_bucket_blkno)
			fc_retain_pin = true;
		else
			fc_retain_pin = false;

		fc_blkno = fc_opaque->hasho_nextblkno;

		/*
		 * 应用删除，推进到下一页，如果需要，写入页面。
		 */
		if (fc_ndeletable > 0)
		{
			/* 直到更改被记录，不要ereport(ERROR) */
			START_CRIT_SECTION();

			PageIndexMultiDelete(fc_page, fc_deletable, fc_ndeletable);
			fc_bucket_dirty = true;

			/*
			 * 如果vacuum从索引页面中删除DEAD元组，
			 * 让我们将页面标记为干净。我们通过清除
			 * LH_PAGE_HAS_DEAD_TUPLES标志来做到这一点。
			 */
			if (fc_tuples_removed && *fc_tuples_removed > 0 &&
				H_HAS_DEAD_TUPLES(fc_opaque))
			{
				fc_opaque->hasho_flag &= ~LH_PAGE_HAS_DEAD_TUPLES;
				fc_clear_dead_marking = true;
			}

			MarkBufferDirty(fc_buf);

			/* XLOG 相关内容 */
			if (RelationNeedsWAL(fc_rel))
			{
				xl_hash_delete fc_xlrec;
				XLogRecPtr	fc_recptr;

				fc_xlrec.clear_dead_marking = fc_clear_dead_marking;
				fc_xlrec.is_primary_bucket_page = (fc_buf == fc_bucket_buf);

				XLogBeginInsert();
				XLogRegisterData((char *) &fc_xlrec, SizeOfHashDelete);

				/*
				 * bucket buffer 需要被注册以确保我们可以
				 * 在重放期间对其获取清理锁。
				 */
				if (!fc_xlrec.is_primary_bucket_page)
					XLogRegisterBuffer(0, fc_bucket_buf, REGBUF_STANDARD | REGBUF_NO_IMAGE);

				XLogRegisterBuffer(1, fc_buf, REGBUF_STANDARD);
				XLogRegisterBufData(1, (char *) fc_deletable,
									fc_ndeletable * sizeof(OffsetNumber));

				fc_recptr = XLogInsert(RM_HASH_ID, XLOG_HASH_DELETE);
				PageSetLSN(BufferGetPage(fc_buf), fc_recptr);
			}

			END_CRIT_SECTION();
		}

		/* 如果没有更多页面可扫描，则退出。 */
		if (!BlockNumberIsValid(fc_blkno))
			break;

		fc_next_buf = _hash_getbuf_with_strategy(fc_rel, fc_blkno, HASH_WRITE,
											  LH_OVERFLOW_PAGE,
											  fc_bstrategy);

		/*
		 * 在获取下一个页面的锁后释放对前一个页面的锁
		 */
		if (fc_retain_pin)
			LockBuffer(fc_buf, BUFFER_LOCK_UNLOCK);
		else
			_hash_relbuf(fc_rel, fc_buf);

		fc_buf = fc_next_buf;
	}

	/*
	 * 锁定 bucket 页面以清除垃圾标志并压缩 bucket。
	 * 如果当前缓冲区与 bucket 缓冲区相同，则我们已经在
	 * bucket 页面上获得了锁。
	 */
	if (fc_buf != fc_bucket_buf)
	{
		_hash_relbuf(fc_rel, fc_buf);
		LockBuffer(fc_bucket_buf, BUFFER_LOCK_EXCLUSIVE);
	}

	/*
	 * 在删除由拆分移动的元组后，从 bucket 中清除垃圾标志。
	 * 我们故意在压缩 bucket 之前清除该标志，
	 * 这样在重启后，清理程序就不应该再次尝试删除通过拆分移动的元组。
	 */
	if (fc_split_cleanup)
	{
		HashPageOpaque fc_bucket_opaque;
		Page		fc_page;

		fc_page = BufferGetPage(fc_bucket_buf);
		fc_bucket_opaque = HashPageGetOpaque(fc_page);

		/* 直到更改被记录，不要ereport(ERROR) */
		START_CRIT_SECTION();

		fc_bucket_opaque->hasho_flag &= ~LH_BUCKET_NEEDS_SPLIT_CLEANUP;
		MarkBufferDirty(fc_bucket_buf);

		/* XLOG 相关内容 */
		if (RelationNeedsWAL(fc_rel))
		{
			XLogRecPtr	fc_recptr;

			XLogBeginInsert();
			XLogRegisterBuffer(0, fc_bucket_buf, REGBUF_STANDARD);

			fc_recptr = XLogInsert(RM_HASH_ID, XLOG_HASH_SPLIT_CLEANUP);
			PageSetLSN(fc_page, fc_recptr);
		}

		END_CRIT_SECTION();
	}

	/*
	 * 如果我们删除了任何东西，尝试紧凑空闲空间。
	 * 对于压缩 bucket，我们必须拥有清理锁，否则可能会影响
	 * 对于一个已经开始的扫描的元组的顺序。
	 */
	if (fc_bucket_dirty && IsBufferCleanupOK(fc_bucket_buf))
		_hash_squeezebucket(fc_rel, fc_cur_bucket, fc_bucket_blkno, fc_bucket_buf,
							fc_bstrategy);
	else
		LockBuffer(fc_bucket_buf, BUFFER_LOCK_UNLOCK);
}
